package coder.taiji.springnetty.utils;

import io.netty.channel.ChannelHandlerContext;

import java.util.concurrent.*;


/**
 * NettyPushMessage 多线程的方式实现同步下发数据。
 * <p>
 * 在线程A sendMessage()发送数据，然后等待结果。EventLoop 线程收到同一个 channel 响应之后调用 complete(T result) 方法
 * 即可使A线程 sendMessage() 返回,并获取到结果 result
 *
 * @param <T>发送线程和EventLoop 线程的结果类型
 * @author helloworldyu
 * @date 2017/6/30
 */

public class NettyPushMessage<T> {
	/**
	 * 报存所有下发了数据，在等待结果的 channel 和 线程间通信机制的关系
	 */
	private ConcurrentHashMap<ChannelHandlerContext, CompletableFuture<T>>
			contextFuture = new ConcurrentHashMap<>();

	/**
	 * 发送消息,等待 EventLoop 线程通过 complete() 调用返回了结果的。
	 *
	 * @param message 消息
	 * @param timeout 超时时间 单位秒
	 * @return 成功执行之后的结果
	 * @throws CancellationException if this future was cancelled
	 * @throws ExecutionException    if this future completed exceptionally
	 * @throws InterruptedException  if the current thread was interrupted
	 *                               while waiting
	 * @throws TimeoutException      if the wait timed out
	 */


	public T sendMessage(ChannelHandlerContext ctx, Object message, long timeout)
			throws CancellationException, InterruptedException, ExecutionException, TimeoutException {
		//判断是否已经有消息在下发了
		boolean exist = isExistPushed(ctx);
		if (exist) {
			throw new RuntimeException("已经有调用阻塞sendMessage() push 数据到 此channle");
		}
		//创建线程间通信变量
		CompletableFuture<T> completableFuture = new CompletableFuture<>();
		//放进map
		contextFuture.put(ctx, completableFuture);
		//发送数据
		ctx.writeAndFlush(message);
		T result;
		try {
			//等待结果
			result = completableFuture.get(timeout, TimeUnit.SECONDS);
		} finally {
			//不管成功与否都得删除
			contextFuture.remove(ctx);
		}

		return result;
	}

	/**
	 * 是否已经有调用阻塞 sendMessage() 向此 ctx 关联的 channle 发送数据
	 *
	 * @param ctx channel 与 当前handler的上下文
	 * @return 此 channel 是否正在下发数据。
	 */
	public boolean isExistPushed(ChannelHandlerContext ctx) {
		return contextFuture.containsKey(ctx);
	}

	/**
	 * 发送数据,不等待结果
	 *
	 * @param message 下发的消息
	 */
	public void sendMessage(ChannelHandlerContext ctx, Object message) {
		ctx.writeAndFlush(message);
	}

	/**
	 * EventLoop 线程收到数据之后，返回结果
	 *
	 * @param ctx    接收数据的 channle 的 ctx
	 * @param result 执行结果
	 * @return false 没有在等待这个通知的线程了(例如等待超时)。 或者通知转换状态为成功失败
	 */
	public boolean complete(ChannelHandlerContext ctx, T result) {
		//通过 channel 的 ctx 找到 CompletableFuture
		CompletableFuture<T> completableFuture = contextFuture.get(ctx);
		if (null == completableFuture) {
			return false;
		} else {
			return completableFuture.complete(result);
		}
	}

	/**
	 * EventLoop 线程收到数据之后，不关心执行的结果
	 *
	 * @param ctx 接收数据的 channle 的 ctx
	 * @return false 没有在等待这个通知的线程了(例如等待超时)。 或者通知转换状态为成功失败
	 */
	public boolean complete(ChannelHandlerContext ctx) {
		CompletableFuture completableFuture = contextFuture.get(ctx);
		if (null == completableFuture) {
			return false;
		}

		return completableFuture.complete(null);
	}
}